#include "Avalon/IO/PipedReader.hpp"
#include "Avalon/IO/BufferReader.hpp"
#include "Avalon/IO/IOException.hpp"
#include "Avalon/IO/PipedWriter.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace std;

PipedReader::PipedReader()
    : m_writeMutex(new mutex()),
      m_result(NULL),
      m_buffer(NULL),
      m_destination(NULL),
      m_readIterator(m_source.GetData()),
      m_readRemaining(0),
      m_writer(NULL) {}

PipedReader::~PipedReader() {
  Break();
}

void PipedReader::Break() {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak(readLock, writeLock);
}

void PipedReader::Break(const exception_ptr& cause) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  m_cause = cause;
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak(readLock, writeLock);
}

void PipedReader::Read(void* destination, int size,
    Out<Async<int> > sizeRead) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  if(m_readRemaining != 0) {
    int copySize = min(m_readRemaining, size);
    memcpy(destination, m_readIterator, copySize);
    m_readIterator += copySize;
    m_readRemaining -= copySize;
    readLock.unlock();
    writeLock.unlock();
    sizeRead->SetResult(copySize);
    return;
  }
  if(m_writer == NULL) {
    readLock.unlock();
    writeLock.unlock();
    if(m_cause == NULL) {
      sizeRead->SetException(IOException("Pipe broken."));
    } else {
      sizeRead->SetException(m_cause);
    }
    return;
  }
  m_result = sizeRead.Get();
  m_buffer = NULL;
  m_destination = destination;
  m_size = size;
}

void PipedReader::Read(Buffer* destination, int size,
    Out<Async<int> > sizeRead) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  if(m_readRemaining != 0) {
    int copySize = min(m_readRemaining, size);
    int copyOffset = destination->GetSize();
    destination->Grow(copySize);
    memcpy(destination->GetMutableData() + copyOffset, m_readIterator,
      copySize);
    m_readIterator += copySize;
    m_readRemaining -= copySize;
    readLock.unlock();
    writeLock.unlock();
    sizeRead->SetResult(copySize);
    return;
  }
  if(m_writer == NULL) {
    readLock.unlock();
    writeLock.unlock();
    if(m_cause == NULL) {
      sizeRead->SetException(IOException("Pipe broken."));
    } else {
      sizeRead->SetException(m_cause);
    }
    return;
  }
  m_result = sizeRead.Get();
  m_buffer = destination;
  m_destination = NULL;
  m_size = size;
}

void PipedReader::Connect(PipedWriter* writer) {
  m_writer = writer;
}

void PipedReader::InternalBreak(unique_lock<mutex>& readLock,
    unique_lock<mutex>& writeLock) {
  m_writer = NULL;
  if(m_result == NULL) {
    return;
  }
  Async<int>* result = m_result;
  m_result = NULL;
  m_buffer = NULL;
  m_destination = NULL;
  readLock.unlock();
  writeLock.unlock();
  if(m_cause == NULL) {
    result->SetException(IOException("Pipe broken."));
  } else {
    result->SetException(m_cause);
  }
}

void PipedReader::Write(const void* data, int size,
    unique_lock<mutex>& readLock, unique_lock<mutex>& writeLock) {
  if(m_result != NULL) {
    int copySize = min(size, m_size);
    void* destination;
    if(m_buffer != NULL) {
      int copyOffset = m_buffer->GetSize();
      m_buffer->Grow(copySize);
      destination = m_buffer->GetMutableData() + copyOffset;
    } else {
      destination = m_destination;
    }
    memcpy(destination, data, copySize);
    if(copySize < size) {
      m_readRemaining = size - copySize;
      m_source.Append(static_cast<const char*>(data) + copySize,
        m_readRemaining);
      m_readIterator = m_source.GetData();
    }
    Async<int>* result = m_result;
    m_result = NULL;
    m_buffer = NULL;
    m_destination = NULL;
    readLock.unlock();
    writeLock.unlock();
    result->SetResult(copySize);
    return;
  }
  ptrdiff_t sizeRead = m_readIterator - m_source.GetData();
  static const int MIN_OVERHEAD = 1024;
  if(sizeRead >= max(m_source.GetSize() / 2, MIN_OVERHEAD)) {
    Buffer newBuffer(m_readIterator, m_readRemaining);
    swap(m_source, newBuffer);
    sizeRead = 0;
  }
  m_source.Append(data, size);
  m_readIterator = m_source.GetData() + sizeRead;
  m_readRemaining += size;
}
